package com.beametl.sample.origin;

import java.sql.ResultSet;

import com.beametl.sample.entity.StaffVO;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.SerializableCoder;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;

/*关系型数据库到文本之间的ETL传输样例*/
public class JDBCToTextSampleOrigin {
    public static void dealWithPipeline(Pipeline pipeline) {

        JdbcIO.DataSourceConfiguration config = JdbcIO.DataSourceConfiguration
                .create("com.ibm.db2.jcc.DB2Driver", "jdbc:db2://10.251.106.56:50400/RBSDBT2").withUsername("xxxxx")
                .withPassword("xxxxx");

        pipeline.apply(JdbcIO.<StaffVO>read().withFetchSize(1000).withDataSourceConfiguration(config)
                .withQuery("select staffid,staffname from baozhang.wf_staff")
                .withRowMapper(new JdbcIO.RowMapper<StaffVO>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public StaffVO mapRow(ResultSet rs) throws Exception {

                        StaffVO staffVO = new StaffVO();
                        staffVO.setStaffid(rs.getString("STAFFID"));
                        staffVO.setStaffname(rs.getString("STAFFNAME"));
                        return staffVO;
                    }
                }).withCoder(SerializableCoder.of(StaffVO.class)))
                .apply("ExtractWords", ParDo.of(new DoFn<StaffVO, String>() {
                    private static final long serialVersionUID = 4502730154830948266L;

                    @DoFn.ProcessElement
                    public void processElement(ProcessContext c) {
                        c.output(c.element().getStaffid() + "," + c.element().getStaffname());
                    }
                })).apply(TextIO.write().to("G:/2.txt"));

    }
}